package com.ehualu.impda.hadoop;

import com.ehualu.impda.kafkasecurity.LoginUtilKafkaHuawei;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;


public class NewProducer3 {
    private static final Logger LOG = LoggerFactory.getLogger(NewProducer3.class);

    public static KafkaProducer<Integer, String> producer;

    public static String topic;

    public static Boolean isAsync;


    public static String bootstrapServers = "bootstrap.servers";

    public static String clientId = "client.id";

    public static String keySerializer = "key.serializer";

    public static String valueSerializer = "value.serializer";

    public static String securityProtocol = "security.protocol";

    public static String saslKerberosServiceName = "sasl.kerberos.service.name";

    public static String kerberosDomainName = "kerberos.domain.name";

    public static int messageNumToSend = 100;

    /**
     * 用户自己申请的机机账号keytab文件名称
     */
    /**
     * 用户自己申请的机机账号keytab文件名称
     */
    private static final String USER_KEYTAB_FILE = "user.keytab";

    /**
     * 用户自己申请的机机账号名称
     */
    private static final String USER_PRINCIPAL = "super_rj";


    /**
     * 新Producer 构造函数
     *
     * @param topicName Topic名称
     */
//    public NewProducer3(String topicName, Boolean asyncEnable) {
//
//
//        KafkaProperties kafkaProc = KafkaProperties.getInstance();
//
//        // Broker地址列表
//        props.put(bootstrapServers, kafkaProc.getValues(bootstrapServers, "37.158.97.137:21005,37.158.97.136:21005"));
//        // 客户端ID
//        props.put(clientId, kafkaProc.getValues(clientId, "DemoProducer"));
//        // Key序列化类
//        props.put(keySerializer,
//                kafkaProc.getValues(keySerializer, "org.apache.kafka.common.serialization.IntegerSerializer"));
//        // Value序列化类
//        props.put(valueSerializer,
//                kafkaProc.getValues(valueSerializer, "org.apache.kafka.common.serialization.StringSerializer"));
//        // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
//        props.put(securityProtocol, kafkaProc.getValues(securityProtocol, "SASL_PLAINTEXT"));
//        // 服务名
//        props.put(saslKerberosServiceName, "kafka");
//        // 域名
//        props.put(kerberosDomainName, kafkaProc.getValues(kerberosDomainName, "hadoop.hadoop.com"));
//
//        producer = new KafkaProducer<Integer, String>(props);
//        topic = topicName;
//        isAsync = asyncEnable;
//    }

    /**
     * 生产者线程执行函数，循环发送消息。
     */
    public void run() {
        LOG.info("New Producer: start.");
        int messageNo = 1;
        // 指定发送多少条消息后sleep1秒
        int intervalMessages = 10;

        while (messageNo <= messageNumToSend) {
            String messageStr = "Message_" + messageNo;
            long startTime = System.currentTimeMillis();

            // 构造消息记录
            ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(topic, messageNo, messageStr);


            if (isAsync) {
                // 异步发送
//                producer.send(record, new DemoCallBack(startTime, messageNo, messageStr));
            } else {
                try {
                    // 同步发送
                    producer.send(record).get();
                } catch (InterruptedException ie) {
                    LOG.info("The InterruptedException occured : {}.", ie);
                } catch (ExecutionException ee) {
                    LOG.info("The ExecutionException occured : {}.", ee);
                }
            }
            messageNo++;

            if (messageNo % intervalMessages == 0) {
                // 每发送intervalMessage条消息sleep1秒
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                LOG.info("The Producer have send {} messages.", messageNo);
            }
        }

    }

    public static void securityPrepare() throws IOException {
        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator;
        String krbFile = filePath + "krb5.conf";
        String userKeyTableFile = filePath + USER_KEYTAB_FILE;

        //windows路径下分隔符替换
        userKeyTableFile = userKeyTableFile.replace("\\", "\\\\");
        krbFile = krbFile.replace("\\", "\\\\");

        LoginUtilKafkaHuawei.setKrb5Config(krbFile);
        LoginUtilKafkaHuawei.setZookeeperServerPrincipal("zookeeper/hadoop.hadoop.com");
        LoginUtilKafkaHuawei.setJaasFile(USER_PRINCIPAL, userKeyTableFile);
    }

    public static Boolean isSecurityModel() {
        Boolean isSecurity = false;
        String krbFilePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator + "kafkaSecurityMode";

        Properties securityProps = new Properties();

        // file does not exist.
        if (!isFileExists(krbFilePath)) {
            return isSecurity;
        }

        try {
            securityProps.load(new FileInputStream(krbFilePath));
            if ("yes".equalsIgnoreCase(securityProps.getProperty("kafka.client.security.mode"))) {
                isSecurity = true;
            }
        } catch (Exception e) {
            LOG.info("The Exception occured : {}.", e);
        }

        return isSecurity;
    }

    /*
     * 判断文件是否存在
     */
    private static boolean isFileExists(String fileName) {
        File file = new File(fileName);

        return file.exists();
    }

    public static void main(String[] args) {

//        System.setProperty("java.security.auth.login.config","E:\\impda\\src\\main\\resources\\jaas.conf");
        Properties props = new Properties();

        KafkaProperties kafkaProc = KafkaProperties.getInstance();

        // Broker地址列表
        props.put(bootstrapServers, kafkaProc.getValues(bootstrapServers, "37.158.97.137:21007,37.158.97.136:21007,37.158.97.138:21007,37.158.97.139:21007,37.158.97.140:21007"));
        // 客户端ID
        props.put(clientId, kafkaProc.getValues(clientId, "DemoProducer"));
        // Key序列化类
        props.put(keySerializer,
                kafkaProc.getValues(keySerializer, "org.apache.kafka.common.serialization.IntegerSerializer"));
        // Value序列化类
        props.put(valueSerializer,
                kafkaProc.getValues(valueSerializer, "org.apache.kafka.common.serialization.StringSerializer"));
        // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
        props.put(securityProtocol, kafkaProc.getValues(securityProtocol, "SASL_PLAINTEXT"));
        // 服务名
        props.put(saslKerberosServiceName, "kafka");
        // 域名
        props.put(kerberosDomainName, kafkaProc.getValues(kerberosDomainName, "hadoop.hadoop.com"));

        producer = new KafkaProducer<Integer, String>(props);
        topic = "TEST01";


//        if (isSecurityModel()) {
            try {
                LOG.info("Securitymode start.");

                //!!注意，安全认证时，需要用户手动修改为自己申请的机机账号
                securityPrepare();



                int messageNo = 1;
                // 指定发送多少条消息后sleep1秒
                int intervalMessages = 10;

                while (messageNo <= 1) {
//                    String messageStr = "Message_" + messageNo;

                    String messageStr="371501_1939729385,1939729385,371501,1,2019-06-03 01:04:19,3715000916,聊城市交运超市门口 礼让行人1,jd01,wd02,鲁PT1910,2,4,K31,H,02,0,http://37.158.66.59:6120/pic?4dd12=sba-z2531145503a4a--dc24551d44a0ei9b3*=6d9*24d1i*s1d=i2p7t=pe*m5i10=906-e99*ac92i5b0ie9=,http://37.158.66.59:6120/pic?=d08i0b8e*9cai20e-699a0a-9dc24551d44a0ei9b3*=6d9*24d1i*s1d=i3p7t=pe*m5i14=-155430-52z131s=8a4bd8,,,,,,,,,,,,,,,,,,";
                    long startTime = System.currentTimeMillis();

                    // 构造消息记录
                    ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(topic, messageStr);




                    try {
                        // 同步发送
                        producer.send(record).get();
                    } catch (InterruptedException ie) {
                        LOG.info("The InterruptedException occured : {}.", ie);
                    } catch (ExecutionException ee) {
                        LOG.info("The ExecutionException occured : {}.", ee);
                    }

                    messageNo++;

//                    if (messageNo % intervalMessages == 0) {
//                        // 每发送intervalMessage条消息sleep1秒
//                        try {
//                            Thread.sleep(1000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
//                        LOG.info("The Producer have send {} messages.", messageNo);
//                    }
                }


            } catch (IOException e) {
                LOG.error("Security prepare failure.");
                LOG.error("The IOException occured.", e);
                return;
            }
            LOG.info("Security prepare success.");
//        }

        // 是否使用异步发送模式
//        final boolean asyncEnable = false;
//        NewProducer3 producerThread = new NewProducer3(KafkaProperties.TOPIC, asyncEnable);
//        producerThread.start();
    }
}
//
//class DemoCallBack implements Callback {
//    private static Logger LOG = LoggerFactory.getLogger(DemoCallBack.class);
//
//    private long startTime;
//
//    private int key;
//
//    private String message;
//
//    public DemoCallBack(long startTime, int key, String message) {
//        this.startTime = startTime;
//        this.key = key;
//        this.message = message;
//    }
//
//    /**
//     * 回调函数，用于处理异步发送模式下，消息发送到服务端后的处理。
//     *
//     * @param metadata  元数据信息
//     * @param exception 发送异常。如果没有错误发生则为Null。
//     */
//    @Override
//    public void onCompletion(RecordMetadata metadata, Exception exception) {
//        long elapsedTime = System.currentTimeMillis() - startTime;
//        if (metadata != null) {
//            LOG.info("message(" + key + ", " + message + ") sent to partition(" + metadata.partition() + "), "
//                    + "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
//        } else if (exception != null) {
//            LOG.error("The Exception occured.", exception);
//        }
//
//    }
//}